/*-------------------------------------------------------------------------
 *
 * deadlock.c
 *	  POSTGRES 死锁检测代码
 *
 * 请参阅 src/backend/storage/lmgr/README 以获取死锁检测和解决算法的描述。
 *
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  src/backend/storage/lmgr/deadlock.c
 *
 *	接口：
 *
 *	DeadLockCheck()
 *	DeadLockReport()
 *	RememberSimpleDeadLock()
 *	InitDeadLockChecking()
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "miscadmin.h"
#include "pg_trace.h"
#include "pgstat.h"
#include "storage/lmgr.h"
#include "storage/proc.h"
#include "utils/memutils.h"


/*
 * 等待图中的一条边。
 *
 * 等待者和阻塞者可能是锁组的成员，也可能不是，但如果
 * 其中任何一个是，它将是该锁组的领导者，而不是锁
 *组的其他成员。组领导者作为整个组的代表，尽管
 *这些特定的进程不必等待。对于给定的锁，等待队列
 *中至少会有一个等待者的锁组成员，也可能有更多。
 */
typedef struct
{
	PGPROC	   *waiter;			/* 等待锁组的领导者 */
	PGPROC	   *blocker;		/* 它正在等待的组的领导者 */
	LOCK	   *lock;			/* 正在等待的锁 */
	int			pred;			/* TopoSort 的工作空间 */
	int			link;			/* TopoSort 的工作空间 */
} EDGE;

/* 锁的等待队列的一个潜在重新排序 */
typedef struct
{
	LOCK	   *lock;			/* 描述的锁的等待队列 */
	PGPROC	  **procs;			/* 新的等待顺序中的 PGPROC * 的数组 */
	int			nProcs;
} WAIT_ORDER;

/*
 * 关于检测到的死锁循环中每个边的信息。 
 * 这用于在失败时打印诊断消息。
 *
 * 注意：因为我们希望在释放锁管理器的分区锁后检查这些信息，
 * 所以不能仅存储 LOCK 和 PGPROC 指针；我们必须提取所有希望打印的信息。
 */
typedef struct
{
	LOCKTAG		locktag;		/* 等待的锁对象的 ID */
	LOCKMODE	lockmode;		/* 我们正在等待的锁的类型 */
	int			pid;			/* 被阻塞的后端的 PID */
} DEADLOCK_INFO;


static bool fc_DeadLockCheckRecurse(PGPROC *fc_proc);
static int	fc_TestConfiguration(PGPROC *fc_startProc);
static bool fc_FindLockCycle(PGPROC *fc_checkProc,
						  EDGE *fc_softEdges, int *fc_nSoftEdges);
static bool fc_FindLockCycleRecurse(PGPROC *fc_checkProc, int fc_depth,
								 EDGE *fc_softEdges, int *fc_nSoftEdges);
static bool fc_FindLockCycleRecurseMember(PGPROC *fc_checkProc,
									   PGPROC *fc_checkProcLeader,
									   int fc_depth, EDGE *fc_softEdges, int *fc_nSoftEdges);
static bool fc_ExpandConstraints(EDGE *fc_constraints, int fc_nConstraints);
static bool fc_TopoSort(LOCK *fc_lock, EDGE *fc_constraints, int fc_nConstraints,
					 PGPROC **fc_ordering);

#ifdef DEBUG_DEADLOCK
static void fc_PrintLockQueue(LOCK *fc_lock, const char *fc_info);
#endif


/*
 * 死锁检测器的工作空间
 */

/* FindLockCycle 的工作空间 */
static PGPROC **visitedProcs;	/* 已访问进程的数组 */
static int	nVisitedProcs;

/* TopoSort 的工作空间 */
static PGPROC **topoProcs;		/* 尚未输出进程的数组 */
static int *beforeConstraints;	/* 剩余的前约束计数 */
static int *afterConstraints;	/* 后约束的列表头 */

/* ExpandConstraints 的输出区域 */
static WAIT_ORDER *waitOrders;	/* 提议的队列重排的数组 */
static int	nWaitOrders;
static PGPROC **waitOrderProcs; /* 等待订单队列内容的空间 */

/* 当前考虑的约束列表 */
static EDGE *curConstraints;
static int	nCurConstraints;
static int	maxCurConstraints;

/* FindLockCycle 结果的存储空间 */
static EDGE *possibleConstraints;
static int	nPossibleConstraints;
static int	maxPossibleConstraints;
static DEADLOCK_INFO *deadlockDetails;
static int	nDeadlockDetails;

/* 找到的任何阻塞自动真空工作进程的 PGPROC 指针 */
static PGPROC *blocking_autovacuum_proc = NULL;


/*
 * InitDeadLockChecking -- 在后端启动期间初始化死锁检查器
 *
 * 这对每个后端进行死锁检查器的初始化；主要是
 * 为 DeadLockCheck 分配工作内存。我们对每个后端执行此操作
 * 是因为让内核执行 copy-on-write
 * 从主进程继承工作空间没有意义。我们希望在启动时分配
 * 空间，因为 (a) 死锁检查器可能会在没有可用内存的情况下被调用，且
 * (b) 检查器通常在信号处理程序内运行，这是一个非常危险的地方进行 palloc。
 */
void InitDeadLockChecking(void)
{
	MemoryContext fc_oldcxt;

	/* 确保分配是永久的 */
	fc_oldcxt = MemoryContextSwitchTo(TopMemoryContext);

	/*
	 * FindLockCycle 至多需要 MaxBackends 条目的 visitedProcs[] 和
	 * deadlockDetails[]。
	 */
	visitedProcs = (PGPROC **) palloc(MaxBackends * sizeof(PGPROC *));
	deadlockDetails = (DEADLOCK_INFO *) palloc(MaxBackends * sizeof(DEADLOCK_INFO));

	/*
	 * TopoSort 最多需要考虑 MaxBackends 等待队列条目，
	 * 它不必与 FindLockCycle 并发运行。
	 */
	topoProcs = visitedProcs;	/* 重新使用此空间 */
	beforeConstraints = (int *) palloc(MaxBackends * sizeof(int));
	afterConstraints = (int *) palloc(MaxBackends * sizeof(int));

	/*
	 * 我们需要考虑最多 MaxBackends/2 个等待队列的重排
	 * (因为至少需要两个等待者在队列中才能创建一个软边)，
	 * 扩展形式的等待队列不能涉及超过 
	 * MaxBackends 的总等待者。
	 */
	waitOrders = (WAIT_ORDER *)
		palloc((MaxBackends / 2) * sizeof(WAIT_ORDER));
	waitOrderProcs = (PGPROC **) palloc(MaxBackends * sizeof(PGPROC *));

	/*
	 * 在一个配置中允许最多 MaxBackends 个不同的约束。 （这
	 * 够吗？ 实际上看起来应该够，但我不太知道如何证明它。
	 * 如果用完，我们可能无法找到一个可行的等待
	 * 队列重排，尽管一个存在。） 注意，这个数字
	 * 限制 DeadLockCheckRecurse 的最大递归深度。将其
	 * 设置得很大可能会导致栈溢出问题。
	 */
	maxCurConstraints = MaxBackends;
	curConstraints = (EDGE *) palloc(maxCurConstraints * sizeof(EDGE));

	/*
	 * 允许最多保存 3*MaxBackends 个约束而无需
	 * 重新运行 TestConfiguration。 （这可能是足够的，
	 * 但如果空间不足，我们可以通过对
	 * TestConfiguration 进行多次运行来重新计算约束列表。） 
	 * possibleConstraints[] 中的最后 MaxBackends 条目被保留
	 * 作为 FindLockCycle 的输出工作空间。
	 */
	maxPossibleConstraints = MaxBackends * 4;
	possibleConstraints =
		(EDGE *) palloc(maxPossibleConstraints * sizeof(EDGE));

	MemoryContextSwitchTo(fc_oldcxt);
}

/*
 * DeadLockCheck -- 检查给定进程的死锁
 *
 * 这段代码查找与给定进程相关的死锁。如果找到任何死锁，
 * 它尝试重新排列锁等待队列以解决死锁。如果无法解决，
 * 则返回 DS_HARD_DEADLOCK ---
 * 调用者期望要中止给定进程的事务。
 *
 * 调用者必须已经锁定了锁表的所有分区。
 *
 * 失败时，死锁的详细信息将记录在 deadlockDetails[] 中，
 * 以便以后由 DeadLockReport() 打印。该活动是独立的
 * 因为 (a) 我们不想在持有所有 LWLocks 的同时
 * 执行此操作，且 (b) 我们通常在信号处理程序内部被调用。
 */
DeadLockState
DeadLockCheck(PGPROC *fc_proc)
{
	int			fc_i,
				fc_j;

	/* 初始化为“无约束” */
	nCurConstraints = 0;
	nPossibleConstraints = 0;
	nWaitOrders = 0;

	/* 初始化为未被 autovacuum 工作线程阻塞 */
	blocking_autovacuum_proc = NULL;

	/* 搜索死锁及可能的解决方案 */
	if (fc_DeadLockCheckRecurse(fc_proc))
	{
		/*
		 * 再次调用 FindLockCycle，以记录正确的
		 * deadlockDetails[] 对于没有重新排列的基本状态。
		 */
		int			fc_nSoftEdges;

		TRACE_POSTGRESQL_DEADLOCK_FOUND();

		nWaitOrders = 0;
		if (!fc_FindLockCycle(fc_proc, possibleConstraints, &fc_nSoftEdges))
			elog(FATAL, "deadlock seems to have disappeared");

		return DS_HARD_DEADLOCK;	/* 找不到非死锁状态 */
	}

	/* 应用任何需要的等待队列重新排列 */
	for (fc_i = 0; fc_i < nWaitOrders; fc_i++)
	{
		LOCK	   *fc_lock = waitOrders[fc_i].lock;
		PGPROC	  **fc_procs = waitOrders[fc_i].procs;
		int			fc_nProcs = waitOrders[fc_i].nProcs;
		PROC_QUEUE *fc_waitQueue = &(fc_lock->waitProcs);

		Assert(fc_nProcs == fc_waitQueue->size);

#ifdef DEBUG_DEADLOCK
		fc_PrintLockQueue(fc_lock, "DeadLockCheck:");
#endif

		/* 重置队列并按所需顺序重新添加进程 */
		ProcQueueInit(fc_waitQueue);
		for (fc_j = 0; fc_j < fc_nProcs; fc_j++)
		{
			SHMQueueInsertBefore(&(fc_waitQueue->links), &(fc_procs[fc_j]->links));
			fc_waitQueue->size++;
		}

#ifdef DEBUG_DEADLOCK
		fc_PrintLockQueue(fc_lock, "rearranged to:");
#endif

		/* 看看锁的任何等待者现在是否可以被唤醒 */
		ProcLockWakeup(GetLocksMethodTable(fc_lock), fc_lock);
	}

	/* 返回代码告诉调用者我们是否必须逃避死锁 */
	if (nWaitOrders > 0)
		return DS_SOFT_DEADLOCK;
	else if (blocking_autovacuum_proc != NULL)
		return DS_BLOCKED_BY_AUTOVACUUM;
	else
		return DS_NO_DEADLOCK;
}

/*
 * 返回阻塞进程的 autovacuum 的 PGPROC。
 *
 * 我们在将其返回后立即重置保存的指针。
 */
PGPROC * GetBlockingAutoVacuumPgproc(void)
{
	PGPROC	   *fc_ptr;

	fc_ptr = blocking_autovacuum_proc;
	blocking_autovacuum_proc = NULL;

	return fc_ptr;
}

/*
 * DeadLockCheckRecurse -- 递归搜索有效排序
 *
 * curConstraints[] 保存当前正在考虑的约束集合
 * 由外部递归级别使用。在此基础上，为此级别检测到的任何循环
 * 添加每个可能的解决方案约束。
 *
 * 如果不存在解决方案，则返回 true。如果可以实现无死锁状态，则返回 false，
 * 此时 waitOrders[] 显示锁等待队列所需的重新排列（如果有的话）。
 */
static bool fc_DeadLockCheckRecurse(PGPROC *fc_proc)
{
	int			fc_nEdges;
	int			fc_oldPossibleConstraints;
	bool		fc_savedList;
	int			fc_i;

	fc_nEdges = fc_TestConfiguration(fc_proc);
	if (fc_nEdges < 0)
		return true;			/* 硬死锁 --- 无解 */
	if (fc_nEdges == 0)
		return false;			/* 找到良好配置 */
	if (nCurConstraints >= maxCurConstraints)
		return true;			/* 活动约束没有空间？ */
	fc_oldPossibleConstraints = nPossibleConstraints;
	if (nPossibleConstraints + fc_nEdges + MaxBackends <= maxPossibleConstraints)
	{
		/* 我们可以将边缘列表保存在 possibleConstraints[] 中 */
		nPossibleConstraints += fc_nEdges;
		fc_savedList = true;
	}
	else
	{
		/* 没有空间；需要即时重新生成边 */
		fc_savedList = false;
	}

	/*
	 * 尝试每个可用的软边作为配置的添加。
	 */
	for (fc_i = 0; fc_i < fc_nEdges; fc_i++)
	{
		if (!fc_savedList && fc_i > 0)
		{
			/* 重新生成可能添加约束的列表 */
			if (fc_nEdges != fc_TestConfiguration(fc_proc))
				elog(FATAL, "inconsistent results during deadlock check");
		}
		curConstraints[nCurConstraints] =
			possibleConstraints[fc_oldPossibleConstraints + fc_i];
		nCurConstraints++;
		if (!fc_DeadLockCheckRecurse(fc_proc))
			return false;		/* 找到一个有效的解决方案！ */
		/* 放弃那条添加的约束，再试一次 */
		nCurConstraints--;
	}
	nPossibleConstraints = fc_oldPossibleConstraints;
	return true;				/* 未找到解决方案 */
}


/*--------------------
 * 测试配置（当前约束集）的有效性。
 *
 * 返回：
 *		0：配置良好（无死锁）
 *	   -1：配置存在硬死锁或不自洽
 *		>0：配置存在一个或多个软死锁
 *
 * 在软死锁的情况下，任意选择一个软循环，并返回其软边的列表，
 * 从 possibleConstraints+nPossibleConstraints 开始。返回值是
 * 软边的数量。
 *--------------------
 */
static int fc_TestConfiguration(PGPROC *fc_startProc)
{
	int			fc_softFound = 0;
	EDGE	   *fc_softEdges = possibleConstraints + nPossibleConstraints;
	int			fc_nSoftEdges;
	int			fc_i;

	/*
	 * 确保我们有足够的空间容纳 FindLockCycle 的输出。
	 */
	if (nPossibleConstraints + MaxBackends > maxPossibleConstraints)
		return -1;

	/*
	 * 将当前约束集扩展为等待顺序。如果约束集不自洽，则失败。
	 */
	if (!fc_ExpandConstraints(curConstraints, nCurConstraints))
		return -1;

	/*
	 * 检查涉及 startProc 或任何在约束中提到的进程的周期。
	 * 我们最后检查 startProc，因为如果它有一个仍需处理的软循环，
	 * 我们希望先处理这个。
	 */
	for (fc_i = 0; fc_i < nCurConstraints; fc_i++)
	{
		if (fc_FindLockCycle(curConstraints[fc_i].waiter, fc_softEdges, &fc_nSoftEdges))
		{
			if (fc_nSoftEdges == 0)
				return -1;		/* 检测到硬死锁 */
			fc_softFound = fc_nSoftEdges;
		}
		if (fc_FindLockCycle(curConstraints[fc_i].blocker, fc_softEdges, &fc_nSoftEdges))
		{
			if (fc_nSoftEdges == 0)
				return -1;		/* 检测到硬死锁 */
			fc_softFound = fc_nSoftEdges;
		}
	}
	if (fc_FindLockCycle(fc_startProc, fc_softEdges, &fc_nSoftEdges))
	{
		if (fc_nSoftEdges == 0)
			return -1;			/* 检测到硬死锁 */
		fc_softFound = fc_nSoftEdges;
	}
	return fc_softFound;
}


/*
 * FindLockCycle -- 死锁周期的基本检查
 *
 * 从给定进程向外扫描，查看是否存在包含该进程的
 * 等待图中的循环。如果找到循环，则返回 true，否则返回 false。
 * 如果找到循环，我们返回循环中包含的“软边”列表（如果有）。
 * 这些边可能通过重新排列等待队列而消除。我们还将
 * deadlockDetails[] 填充有关检测到的循环的信息；这个信息
 * 不被死锁算法本身使用，仅用于在失败后打印有用的信息。
 *
 * 由于我们需要能够检查在等待队列重新排列后存在的假设配置，
 * 因此该例程关注于 waitOrders[] 中假设的队列顺序表。这些顺序将
 * 被优先相信而不是锁表中看到的实际顺序。
 */
static bool fc_FindLockCycle(PGPROC *fc_checkProc,
			  EDGE *fc_softEdges,	/* 输出参数 */
			  int *fc_nSoftEdges)	/* 输出参数 */
{
	nVisitedProcs = 0;
	nDeadlockDetails = 0;
	*fc_nSoftEdges = 0;
	return fc_FindLockCycleRecurse(fc_checkProc, 0, fc_softEdges, fc_nSoftEdges);
}

static bool fc_FindLockCycleRecurse(PGPROC *fc_checkProc,
					 int fc_depth,
					 EDGE *fc_softEdges,	/* 输出参数 */
					 int *fc_nSoftEdges)	/* 输出参数 */
{
	int			fc_i;
	dlist_iter	fc_iter;

	/*
	 * 如果这个进程是锁组成员，则检查领导者。(注意
	 * 我们可能是领导者，在这种情况下这是一个无操作。)
	 */
	if (fc_checkProc->lockGroupLeader != NULL)
		fc_checkProc = fc_checkProc->lockGroupLeader;

	/*
	 * 我们是否已经见过这个进程？
	 */
	for (fc_i = 0; fc_i < nVisitedProcs; fc_i++)
	{
		if (visitedProcs[fc_i] == fc_checkProc)
		{
			/* 如果我们回到起点，我们就有一个死锁循环 */
			if (fc_i == 0)
			{
				/*
				 * 记录循环的总长度 --- 外层级别现在将填充
				 * deadlockDetails[]
				 */
				Assert(fc_depth <= MaxBackends);
				nDeadlockDetails = fc_depth;

				return true;
			}

			/*
			 * 否则，我们有一个循环，但它不包括起始点，
			 * 所以说“没有死锁”。
			 */
			return false;
		}
	}
	/* 标记proc为已查看 */
	Assert(nVisitedProcs < MaxBackends);
	visitedProcs[nVisitedProcs++] = fc_checkProc;

	/*
	 * 如果进程在等待，则存在一个指向每个阻塞它的进程的
	 * 外发等待边。
	 */
	if (fc_checkProc->links.next != NULL && fc_checkProc->waitLock != NULL &&
		fc_FindLockCycleRecurseMember(fc_checkProc, fc_checkProc, fc_depth, fc_softEdges,
								   fc_nSoftEdges))
		return true;

	/*
	 * 如果进程没有在等待，如果它是锁组的一部分，
	 * 仍然可能存在外发等待边，因为锁组的其他成员可能在
	 * 等待，即使该进程没有。 (给定锁组{A1, A2}和{B1, B2}，
	 * 如果A1等待B1而B2等待A2，
	 * 那就是死锁，即使B1和A2都没有等待任何东西。)
	 */
	dlist_foreach(fc_iter, &fc_checkProc->lockGroupMembers)
	{
		PGPROC	   *fc_memberProc;

		fc_memberProc = dlist_container(PGPROC, lockGroupLink, fc_iter.cur);

		if (fc_memberProc->links.next != NULL && fc_memberProc->waitLock != NULL &&
			fc_memberProc != fc_checkProc &&
			fc_FindLockCycleRecurseMember(fc_memberProc, fc_checkProc, fc_depth, fc_softEdges,
									   fc_nSoftEdges))
			return true;
	}

	return false;
}

static bool fc_FindLockCycleRecurseMember(PGPROC *fc_checkProc,
						   PGPROC *fc_checkProcLeader,
						   int fc_depth,
						   EDGE *fc_softEdges, /* 输出参数 */
						   int *fc_nSoftEdges) /* 输出参数 */
{
	PGPROC	   *fc_proc;
	LOCK	   *fc_lock = fc_checkProc->waitLock;
	PROCLOCK   *fc_proclock;
	SHM_QUEUE  *fc_procLocks;
	LockMethod	fc_lockMethodTable;
	PROC_QUEUE *fc_waitQueue;
	int			fc_queue_size;
	int			fc_conflictMask;
	int			fc_i;
	int			fc_numLockModes,
				fc_lm;

	/*
	 * 关系扩展锁永远不会参与实际的死锁循环。
	 * 参见LockAcquireExtended中的断言。
	 * 因此，检查来自它的等待边没有任何优势。
	 */
	if (LOCK_LOCKTAG(*fc_lock) == LOCKTAG_RELATION_EXTEND)
		return false;

	fc_lockMethodTable = GetLocksMethodTable(fc_lock);
	fc_numLockModes = fc_lockMethodTable->numLockModes;
	fc_conflictMask = fc_lockMethodTable->conflictTab[fc_checkProc->waitLockMode];

	/*
	 * 扫描已经持有冲突锁的进程。这些是在等待图中的“硬”
	 * 边。
	 */
	fc_procLocks = &(fc_lock->procLocks);

	fc_proclock = (PROCLOCK *) SHMQueueNext(fc_procLocks, fc_procLocks,
										 offsetof(PROCLOCK, lockLink));

	while (fc_proclock)
	{
		PGPROC	   *fc_leader;

		fc_proc = fc_proclock->tag.myProc;
		fc_leader = fc_proc->lockGroupLeader == NULL ? fc_proc : fc_proc->lockGroupLeader;

		/* 一个进程永远不会阻塞自己或任何其他锁组成员 */
		if (fc_leader != fc_checkProcLeader)
		{
			for (fc_lm = 1; fc_lm <= fc_numLockModes; fc_lm++)
			{
				if ((fc_proclock->holdMask & LOCKBIT_ON(fc_lm)) &&
					(fc_conflictMask & LOCKBIT_ON(fc_lm)))
				{
					/* 此进程硬性阻塞checkProc */
					if (fc_FindLockCycleRecurse(fc_proc, fc_depth + 1,
											 fc_softEdges, fc_nSoftEdges))
					{
						/* 填充deadlockDetails[] */
						DEADLOCK_INFO *fc_info = &deadlockDetails[fc_depth];

						fc_info->locktag = fc_lock->tag;
						fc_info->lockmode = fc_checkProc->waitLockMode;
						fc_info->pid = fc_checkProc->pid;

						return true;
					}

					/*
					 * 这里没有死锁，但看看这个进程是否是
					 * 直接硬性阻塞我们自身进程的自动清理进程。
					 * 如果是，报告它，以便调用者在适当的时候
					 * 可以向其发送取消信号。如果有多个这样的
					 * 进程，将不确定报告哪个。
					 *
					 * 我们不处理间接阻塞我们的自动清理进程；
					 * 由直接被阻塞的进程采取行动。这条
					 * 规则简化了对行为的理解，并确保
					 * 一个自动清理进程不会在小于
					 * deadlock_timeout宽限期的情况下被取消。
					 *
					 * 注意，我们在没有任何锁定的情况下读取
					 * statusFlags。这只适用于检查PROC_IS_AUTOVACUUM
					 * 标志，因为该标志在进程启动时设置并且
					 * 从不重置。其他地方有逻辑来避免取消一个
					 * 正在防止XID环绕问题的自动清理进程
					 * （这需要读取不同的statusFlags位），
					 * 但是我们在这里不这样做以避免获取
					 * ProcArrayLock。
					 */
					if (fc_checkProc == MyProc &&
						fc_proc->statusFlags & PROC_IS_AUTOVACUUM)
						blocking_autovacuum_proc = fc_proc;

					/* 我们完成了对这个proclock的查看 */
					break;
				}
			}
		}

		fc_proclock = (PROCLOCK *) SHMQueueNext(fc_procLocks, &fc_proclock->lockLink,
											 offsetof(PROCLOCK, lockLink));
	}

	/*
	 * 扫描在锁的等待队列中比这个进程更早的进程。
	 * 那些具有冲突请求的也会软性阻塞这个进程。
	 * 这必须在硬性阻塞搜索之后完成，因为如果另一个进程
	 * 同时硬性和软性阻塞这个进程，我们希望称之为硬边。
	 *
	 * 如果有提议的重新排序锁的等待顺序，使用那个
	 * 而不是当前的等待顺序。
	 */
	for (fc_i = 0; fc_i < nWaitOrders; fc_i++)
	{
		if (waitOrders[fc_i].lock == fc_lock)
			break;
	}

	if (fc_i < nWaitOrders)
	{
		/* 使用给定的假设等待队列顺序 */
		PGPROC	  **fc_procs = waitOrders[fc_i].procs;

		fc_queue_size = waitOrders[fc_i].nProcs;

		for (fc_i = 0; fc_i < fc_queue_size; fc_i++)
		{
			PGPROC	   *fc_leader;

			fc_proc = fc_procs[fc_i];
			fc_leader = fc_proc->lockGroupLeader == NULL ? fc_proc :
				fc_proc->lockGroupLeader;

			/*
			 * TopoSort将始终返回一个顺序，其中组成员
			 * 在等待队列中彼此相邻（请参阅其中的注释）。
			 * 因此，一旦我们到达与checkProc处于同一锁
			 * 组的进程，我们就知道已经找到了位于
			 * checkProcLeader领导的锁组任何成员之前的所有冲突。
			 */
			if (fc_leader == fc_checkProcLeader)
				break;

			/* 这个请求是否有冲突？ */
			if ((LOCKBIT_ON(fc_proc->waitLockMode) & fc_conflictMask) != 0)
			{
				/* 此进程软性阻塞checkProc */
				if (fc_FindLockCycleRecurse(fc_proc, fc_depth + 1,
										 fc_softEdges, fc_nSoftEdges))
				{
					/* 填充deadlockDetails[] */
					DEADLOCK_INFO *fc_info = &deadlockDetails[fc_depth];

					fc_info->locktag = fc_lock->tag;
					fc_info->lockmode = fc_checkProc->waitLockMode;
					fc_info->pid = fc_checkProc->pid;

					/*
					 * 将这个边添加到循环中的软边列表
					 */
					Assert(*fc_nSoftEdges < MaxBackends);
					fc_softEdges[*fc_nSoftEdges].waiter = fc_checkProcLeader;
					fc_softEdges[*fc_nSoftEdges].blocker = fc_leader;
					fc_softEdges[*fc_nSoftEdges].lock = fc_lock;
					(*fc_nSoftEdges)++;
					return true;
				}
			}
		}
	}
	else
	{
		PGPROC	   *fc_lastGroupMember = NULL;

		/* 使用真实的锁等待队列顺序 */
		fc_waitQueue = &(fc_lock->waitProcs);

		/*
		 * 找到在等待队列中存在的锁组的最后一个成员。
		 * 这个之后的任何东西都不是软锁冲突。
		 * 如果没有使用组锁定，我们立刻就知道我们要
		 * 查找哪个进程，但否则我们必须搜索等待队列
		 * 找到最后一个实际存在的进程。
		 */
		if (fc_checkProc->lockGroupLeader == NULL)
			fc_lastGroupMember = fc_checkProc;
		else
		{
			fc_proc = (PGPROC *) fc_waitQueue->links.next;
			fc_queue_size = fc_waitQueue->size;
			while (fc_queue_size-- > 0)
			{
				if (fc_proc->lockGroupLeader == fc_checkProcLeader)
					fc_lastGroupMember = fc_proc;
				fc_proc = (PGPROC *) fc_proc->links.next;
			}
			Assert(fc_lastGroupMember != NULL);
		}

		/*
		 * 好的，现在重新扫描（或扫描）队列以识别软冲突。
		 */
		fc_queue_size = fc_waitQueue->size;
		fc_proc = (PGPROC *) fc_waitQueue->links.next;
		while (fc_queue_size-- > 0)
		{
			PGPROC	   *fc_leader;

			fc_leader = fc_proc->lockGroupLeader == NULL ? fc_proc :
				fc_proc->lockGroupLeader;

			/* 当我们到达目标proc时完成 */
			if (fc_proc == fc_lastGroupMember)
				break;

			/* 这个请求是否有冲突？ */
			if ((LOCKBIT_ON(fc_proc->waitLockMode) & fc_conflictMask) != 0 &&
				fc_leader != fc_checkProcLeader)
			{
				/* 此进程软性阻塞checkProc */
				if (fc_FindLockCycleRecurse(fc_proc, fc_depth + 1,
										 fc_softEdges, fc_nSoftEdges))
				{
					/* 填充deadlockDetails[] */
					DEADLOCK_INFO *fc_info = &deadlockDetails[fc_depth];

					fc_info->locktag = fc_lock->tag;
					fc_info->lockmode = fc_checkProc->waitLockMode;
					fc_info->pid = fc_checkProc->pid;

					/*
					 * 将这个边添加到循环中的软边列表
					 */
					Assert(*fc_nSoftEdges < MaxBackends);
					fc_softEdges[*fc_nSoftEdges].waiter = fc_checkProcLeader;
					fc_softEdges[*fc_nSoftEdges].blocker = fc_leader;
					fc_softEdges[*fc_nSoftEdges].lock = fc_lock;
					(*fc_nSoftEdges)++;
					return true;
				}
			}

			fc_proc = (PGPROC *) fc_proc->links.next;
		}
	}

	/*
	 * 这里没有检测到冲突。
	 */
	return false;
}


/*
 * ExpandConstraints -- 将约束列表扩展为一组
 *		特定的新排序，以影响等待队列
 *
 * 输入是要反转的软边列表。输出是一个包含
 * nWaitOrders WAIT_ORDER结构的列表，
 * 在waitOrders[]中，PGPROC数组工作空间在
 * waitOrderProcs[]中。
 *
 * 如果能够构建满足所有
 * 约束的排序，则返回true，如果不能（存在矛盾约束），
 * 则返回false。
 */
static bool fc_ExpandConstraints(EDGE *fc_constraints,
				  int fc_nConstraints)
{
	int			fc_nWaitOrderProcs = 0;
	int			fc_i,
				fc_j;

	nWaitOrders = 0;

	/*
	 * 反向扫描约束列表。 这是因为最近添加的约束是
	 * 唯一可能失败的，因此我们希望首先测试它的
	 * 不一致性。
	 */
	for (fc_i = fc_nConstraints; --fc_i >= 0;)
	{
		LOCK	   *fc_lock = fc_constraints[fc_i].lock;

		
/* 我们已经为这个锁创建了列表吗？ */
		for (fc_j = nWaitOrders; --fc_j >= 0;)
		{
			if (waitOrders[fc_j].lock == fc_lock)
				break;
		}
		if (fc_j >= 0)
			continue;
		/* 不，没有，所以分配一个新列表 */
		waitOrders[nWaitOrders].lock = fc_lock;
		waitOrders[nWaitOrders].procs = waitOrderProcs + fc_nWaitOrderProcs;
		waitOrders[nWaitOrders].nProcs = fc_lock->waitProcs.size;
		fc_nWaitOrderProcs += fc_lock->waitProcs.size;
		Assert(fc_nWaitOrderProcs <= MaxBackends);

		/*
		 * 进行拓扑排序。拓扑排序在此之后不需要检查约束
		 * ，因为它们必须针对不同的锁。
		 */
		if (!fc_TopoSort(fc_lock, fc_constraints, fc_i + 1,
					  waitOrders[nWaitOrders].procs))
			return false;
		nWaitOrders++;
	}
	return true;
}


/* 
 * TopoSort -- 等待队列的拓扑排序
 *
 * 生成一个锁的等待队列的重排序，该重排序满足关于某些进程在其他进程之前的约束。 (每个约束都是部分排序的事实。) 
 * 最小化为了实现部分排序而不需要的队列重排。
 *
 * 这比例如Knuth的第一卷中展示的拓扑排序算法简单得多且速度较慢。 然而，Knuth的方法并不试图最小化对现有顺序的损害。 
 * 在实践中，我们不太可能处理超过几个约束，因此算法的明显缓慢并不会真正成为问题。
 *
 * 初始队列顺序直接取自锁的等待队列。 
 * 输出是一个PGPROC指针的数组，长度等于锁的等待队列长度（调用者负责提供这个空间）。 
 * 部分顺序由EDGE结构的数组指定。 每个EDGE都是我们需要反转的，因此“等待者”必须出现在输出数组中的“阻塞者”之前。 
 * EDGE数组可能还包含与其他锁相关的边；这些应该被忽略。
 *
 * 如果能够构建满足所有约束的排序，则返回true，反之则返回false（存在矛盾的约束）。
 */
static bool fc_TopoSort(LOCK *fc_lock,
		 EDGE *fc_constraints,
		 int fc_nConstraints,
		 PGPROC **fc_ordering)		/* 输出参数 */
{
	PROC_QUEUE *fc_waitQueue = &(fc_lock->waitProcs);
	int			fc_queue_size = fc_waitQueue->size;
	PGPROC	   *fc_proc;
	int			fc_i,
				fc_j,
				fc_jj,
				fc_k,
				fc_kk,
				fc_last;

	/* 首先，填充topoProcs[]数组，按当前顺序排列进程 */
	fc_proc = (PGPROC *) fc_waitQueue->links.next;
	for (fc_i = 0; fc_i < fc_queue_size; fc_i++)
	{
		topoProcs[fc_i] = fc_proc;
		fc_proc = (PGPROC *) fc_proc->links.next;
	}

	/*
	 * 扫描约束，对于数组中的每个进程，生成一个关于它必须在其他某个进程之前的约束数量，
	 * 加上一个关于它必须在其他某个进程之后的约束的列表。 
	 * j'th进程的计数存储在beforeConstraints[j]中，它的列表头存储在afterConstraints[j]中。 
	 * 每个约束在constraints[i].link中存储其列表链接（注意任何约束只会在一个列表中）。 
	 * 第i个约束的before-proc的数组索引存储在constraints[i].pred中。
	 *
	 * 注意，并不是每个约束都一定影响这个特定的等待队列。
	 * 在组锁之前，一个进程最多只能等待一个锁。 
	 * 但是一个锁组可以等待零个、一个或多个锁。 
	 * 由于topoProcs[]是实际等待的进程的数组，而constraints[]是组领导者的数组，我们必须针对每个约束扫描topoProcs[]，
	 * 检查该组的等待者和阻塞者是否都存在。如果是，约束与此等待队列相关；如果不是，则无关。
	 */
	MemSet(beforeConstraints, 0, fc_queue_size * sizeof(int));
	MemSet(afterConstraints, 0, fc_queue_size * sizeof(int));
	for (fc_i = 0; fc_i < fc_nConstraints; fc_i++)
	{
		/*
		 * 找到一个在锁队列上并且是等待锁组的一部分的代表进程。 
		 * 这可能是领导者，也可能不是，也可能根本不在等待中。 
		 * 如果队列中有其他在同一锁组中的进程，将它们的beforeConstraints数量设置为-1，以指示它们应该与其组员一起输出而不是单独考虑。
		 *
		 * 在这个循环及其下面类似的循环中，关键是我们必须始终选择同一锁组中的相同代表成员，
		 * 这样所有约束都与同一个进程相关，并且-1仅与非代表成员相关。
		 * 我们选择topoProcs数组中的最后一个。
		 */
		fc_proc = fc_constraints[fc_i].waiter;
		Assert(fc_proc != NULL);
		fc_jj = -1;
		for (fc_j = fc_queue_size; --fc_j >= 0;)
		{
			PGPROC	   *fc_waiter = topoProcs[fc_j];

			if (fc_waiter == fc_proc || fc_waiter->lockGroupLeader == fc_proc)
			{
				Assert(fc_waiter->waitLock == fc_lock);
				if (fc_jj == -1)
					fc_jj = fc_j;
				else
				{
					Assert(beforeConstraints[fc_j] <= 0);
					beforeConstraints[fc_j] = -1;
				}
			}
		}

		/* 如果没有匹配的等待者，约束与此锁无关。 */
		if (fc_jj < 0)
			continue;

		/*
		 * 类似地，找到一个在锁队列上并等待阻塞锁组的代表进程。 
		 * 同样，这可以是领导者，但不需要是。
		 */
		fc_proc = fc_constraints[fc_i].blocker;
		Assert(fc_proc != NULL);
		fc_kk = -1;
		for (fc_k = fc_queue_size; --fc_k >= 0;)
		{
			PGPROC	   *fc_blocker = topoProcs[fc_k];

			if (fc_blocker == fc_proc || fc_blocker->lockGroupLeader == fc_proc)
			{
				Assert(fc_blocker->waitLock == fc_lock);
				if (fc_kk == -1)
					fc_kk = fc_k;
				else
				{
					Assert(beforeConstraints[fc_k] <= 0);
					beforeConstraints[fc_k] = -1;
				}
			}
		}

		/* 如果没有匹配的阻塞者，约束与此锁无关。 */
		if (fc_kk < 0)
			continue;

		Assert(beforeConstraints[fc_jj] >= 0);
		beforeConstraints[fc_jj]++;	/* 等待者必须在前 */
		/* 将此约束添加到阻塞者的after-constraints列表中 */
		fc_constraints[fc_i].pred = fc_jj;
		fc_constraints[fc_i].link = afterConstraints[fc_kk];
		afterConstraints[fc_kk] = fc_i + 1;
	}

	/*--------------------
	 * 现在反向扫描topoProcs数组。 在每一步，输出没有剩余before-constraints的最后一个进程以及同一锁组的任何其他成员；
	 * 然后减少与其约束的每个进程的beforeConstraints计数。
	 * i = 我们这次要输出的ordering[]条目的索引
	 * j = topoProcs[]的搜索索引
	 * k = 用于扫描进程j的约束列表的临时变量
	 * last = topoProcs中最后一个非空索引（避免冗余搜索）
	 *--------------------
	 */
	fc_last = fc_queue_size - 1;
	for (fc_i = fc_queue_size - 1; fc_i >= 0;)
	{
		int			fc_c;
		int			fc_nmatches = 0;

		/* 找到下一个候选输出 */
		while (topoProcs[fc_last] == NULL)
			fc_last--;
		for (fc_j = fc_last; fc_j >= 0; fc_j--)
		{
			if (topoProcs[fc_j] != NULL && beforeConstraints[fc_j] == 0)
				break;
		}

		/* 如果没有可用候选，拓扑排序失败 */
		if (fc_j < 0)
			return false;

		
/*
		 * 输出锁组中的所有内容。没有必要输出一个排序，
		 * 如果同一锁组的成员在等待队列上不连续：如果在两个
		 * 属于同一组的请求之间有其他等待者，那么要么它与
		 * 这两个请求都有冲突，显然不是解决方案；要么它与
		 * 至多一个请求冲突，因此与组成员连续的排序同构。
		 */
		fc_proc = topoProcs[fc_j];
		if (fc_proc->lockGroupLeader != NULL)
			fc_proc = fc_proc->lockGroupLeader;
		Assert(fc_proc != NULL);
		for (fc_c = 0; fc_c <= fc_last; ++fc_c)
		{
			if (topoProcs[fc_c] == fc_proc || (topoProcs[fc_c] != NULL &&
										 topoProcs[fc_c]->lockGroupLeader == fc_proc))
			{
				fc_ordering[fc_i - fc_nmatches] = topoProcs[fc_c];
				topoProcs[fc_c] = NULL;
				++fc_nmatches;
			}
		}
		Assert(fc_nmatches > 0);
		fc_i -= fc_nmatches;

		/* 更新其前置项的beforeConstraints计数 */
		for (fc_k = afterConstraints[fc_j]; fc_k > 0; fc_k = fc_constraints[fc_k - 1].link)
			beforeConstraints[fc_constraints[fc_k - 1].pred]--;
	}

	/* 完成 */
	return true;
}

#ifdef DEBUG_DEADLOCK
static void fc_PrintLockQueue(LOCK *fc_lock, const char *fc_info)
{
	PROC_QUEUE *fc_waitQueue = &(fc_lock->waitProcs);
	int			fc_queue_size = fc_waitQueue->size;
	PGPROC	   *fc_proc;
	int			fc_i;

	printf("%s lock %p queue ", fc_info, fc_lock);
	fc_proc = (PGPROC *) fc_waitQueue->links.next;
	for (fc_i = 0; fc_i < fc_queue_size; fc_i++)
	{
		printf(" %d", fc_proc->pid);
		fc_proc = (PGPROC *) fc_proc->links.next;
	}
	printf("\n");
	fflush(stdout);
}
#endif

/*
 * 报告检测到的死锁，附带可用的详细信息。
 */
void DeadLockReport(void)
{
	StringInfoData fc_clientbuf;	/* 客户端的错误详细信息 */
	StringInfoData fc_logbuf;		/* 服务器日志的错误详细信息 */
	StringInfoData fc_locktagbuf;
	int			fc_i;

	initStringInfo(&fc_clientbuf);
	initStringInfo(&fc_logbuf);
	initStringInfo(&fc_locktagbuf);

	/* 生成发送给客户端的“等待”行 */
	for (fc_i = 0; fc_i < nDeadlockDetails; fc_i++)
	{
		DEADLOCK_INFO *fc_info = &deadlockDetails[fc_i];
		int			fc_nextpid;

		/* 最后一个进程等待第一个进程... */
		if (fc_i < nDeadlockDetails - 1)
			fc_nextpid = fc_info[1].pid;
		else
			fc_nextpid = deadlockDetails[0].pid;

		/* 重置锁标记缓冲区以保存下一个对象描述 */
		resetStringInfo(&fc_locktagbuf);

		DescribeLockTag(&fc_locktagbuf, &fc_info->locktag);

		if (fc_i > 0)
			appendStringInfoChar(&fc_clientbuf, '\n');

		appendStringInfo(&fc_clientbuf,
						 _("Process %d waits for %s on %s; blocked by process %d."),
						 fc_info->pid,
						 GetLockmodeName(fc_info->locktag.locktag_lockmethodid,
										 fc_info->lockmode),
						 fc_locktagbuf.data,
						 fc_nextpid);
	}

	/* 为服务器重复上述所有内容... */
	appendBinaryStringInfo(&fc_logbuf, fc_clientbuf.data, fc_clientbuf.len);

	/* ...并添加关于查询字符串的信息 */
	for (fc_i = 0; fc_i < nDeadlockDetails; fc_i++)
	{
		DEADLOCK_INFO *fc_info = &deadlockDetails[fc_i];

		appendStringInfoChar(&fc_logbuf, '\n');

		appendStringInfo(&fc_logbuf,
						 _("Process %d: %s"),
						 fc_info->pid,
						 pgstat_get_backend_current_activity(fc_info->pid, false));
	}

	pgstat_report_deadlock();

	ereport(ERROR,
			(errcode(ERRCODE_T_R_DEADLOCK_DETECTED),
			 errmsg("deadlock detected"),
			 errdetail_internal("%s", fc_clientbuf.data),
			 errdetail_log("%s", fc_logbuf.data),
			 errhint("See server log for query details.")));
}

/*
 * RememberSimpleDeadLock：为DeadLockReport设置信息，当
 * ProcSleep检测到简单（双向）死锁时。proc1希望阻塞以获取锁模式
 * ，但proc2已经在等待，并且会被proc1阻塞。
 */
void RememberSimpleDeadLock(PGPROC *fc_proc1,
					   LOCKMODE fc_lockmode,
					   LOCK *fc_lock,
					   PGPROC *fc_proc2)
{
	DEADLOCK_INFO *fc_info = &deadlockDetails[0];

	fc_info->locktag = fc_lock->tag;
	fc_info->lockmode = fc_lockmode;
	fc_info->pid = fc_proc1->pid;
	fc_info++;
	fc_info->locktag = fc_proc2->waitLock->tag;
	fc_info->lockmode = fc_proc2->waitLockMode;
	fc_info->pid = fc_proc2->pid;
	nDeadlockDetails = 2;
}
